/*
 * ActualNode class creates an actual node that is bound to the registry.
 * 
 * Spring 2013 TCSS 558 Team 6
 */

package node;

import message.*;

import java.io.IOException;
import java.io.Serializable;

import java.rmi.AccessException;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Lock;
import java.util.zip.DataFormatException;

import channel.ActualChannel;

import log.Log;

/**
 * 
 * ActualNode is a node bound to the registry. It simulates virtual nodes and is
 * able to communicate with other actual nodes and virtual nodes.
 * 
 * @author 558 team 6
 * @version Spring 2013
 * 
 */
public class ActualNode implements Node
{
  /**
   * 2^ my_M_value is the total number of all the nodes.
   */
  private final int my_m_value;

  /**
   * It is the identified ID for one node.
   */
  private final int my_id;

  /**
   * system size.
   */
  private static int SYSTEMSIZE;

  /**
   * The virtual nodes that this actual node simulates.
   */
  private final Map<Integer, VirtualNode> my_virtual_nodes =
      new HashMap<Integer, VirtualNode>();

  /**
   * Stores results for requests.
   */
  private final Map<String, Message> my_results_table;

  /**
   * A list of all the successor actual nodes.
   */
  private final List<Integer> my_actual_list = new LinkedList<Integer>();

  /**
   * The mapping between the connection channels between this node and its
   * neighbors.
   */
  private final Map<Integer, ActualChannel> my_channels =
      new HashMap<Integer, ActualChannel>();

  /**
   * The actual node id of the successor node.
   */
  private Node my_successor;

  /**
   * The actual node id that if this node crashes or leaves, it will take all
   * its virtual nodes. (It is the predecessor node in Chord paper)
   */
  private Node my_predecessor;
/**
 * 
 */
  private final Object my_message_monitor = new Object();

/**
 * 
 */

  private final ExecutorService my_thrd_pool;

  /**
   * out is the output stream.
   */
  private final Log my_out;
  
  /**
  * registry.
  */
  private final Registry my_registry;


  /**
   * Constructor of the ActualNode. It assigns the ID to node.
   * 
   * @param the_id is the ID number.
   * @param the_m determines the total nodes that is 2 to the m.
   * @param the_registry is the registry.
   * @throws IOException if there are problems opens the file or creates the
   *           output stream.
   */
  public ActualNode(final int the_id, final int the_m, final Registry the_registry)
      throws IOException
  {
    my_m_value = the_m;
    my_registry = the_registry;
    my_id = the_id;
    my_out = new Log("Log-" + the_id + ".txt");
    // uses the thread pool as the blocking queue.
    my_thrd_pool = Executors.newFixedThreadPool(1);
    my_results_table = Collections.synchronizedMap(new HashMap<String, Message>());
    SYSTEMSIZE = (1 << my_m_value);
  }

  /**
   * Gets the predecessor node index that will take all its virtual nodes when
   * this actual node crashes.
   * 
   * @param the_index is the index of this node.
   * @param the_length is the length of all the actual nodes.
   * @return the predecessor index.
   */
  public String computePredecessor(final int the_index, final int the_length,
                                   final Object[] the_nodes)
  {
    int result = 0;
    if (the_index == 0)
    {
      result = the_length - 1;
    }
    else
    {
      result = the_index - 1;
    }
    return Conversion.IntToString((int) the_nodes[result]);
  }

  /**
   * Gets the successor node index.
   * 
   * @param the_index is the index of this node.
   * @param the_length is the length of all the actual nodes.
   * @return the predecessor id.
   */
  public String computeSuccessor(final int the_index, final int the_length,
                                 final Object[] the_nodes)
  {
    int result = 0;
    if (the_index >= the_length - 1)
    {
      result = 0;
    }
    else
    {
      result = the_index + 1;
    }
    return Conversion.IntToString((int) the_nodes[result]);
  }

  /**
   * According to the information in the registry, it initializes the list of
   * Actual Node, list of Virtual Node, Actual Channel for this node.
   * 
   * @throws RemoteException if there is a problem completing the method call.
   * @throws InterruptedException
   */
  public void initialize() throws RemoteException
  {
    final String[] actual_nodes = getRegistry().list();
    // sorts the array
    final SortedSet<Integer> sorted_nodes = new TreeSet<Integer>();
    for (String s : actual_nodes)
    {
      sorted_nodes.add(Conversion.StringToInt(s));
    }
    initialNodeList(sorted_nodes.toArray());
    System.out.println("Node " + getMyID() + ": successor = " + my_successor.getMyID() +
                       " | predecesor = " + my_predecessor.getMyID());
    initializeVirtualNodes();
    initializeChannel(sorted_nodes.toArray());
    log();
  }

 /**
  * 
  */
  public void reinitialize() throws RemoteException
  {
    System.out.println("Node " + getMyID() + " is reinitializing its" +
                       " successor and predecessors");
    final String[] actual_nodes = getRegistry().list();
    // sorts the array
    final SortedSet<Integer> sorted_nodes = new TreeSet<Integer>();
    for (String s : actual_nodes)
    {
      sorted_nodes.add(Conversion.StringToInt(s));
    }
    initialNodeList(sorted_nodes.toArray());
    System.out.println("Node " + getMyID() + " new successor = " + my_successor.getMyID() +
                       " | new predecesor = " + my_predecessor.getMyID());
    // TODO should it initialize channel for successor and predecessor?
    initializeChannel(sorted_nodes.toArray());
    System.out.println("Node " + getMyID() + " has updated its channels");
  }

  /**
   * Initializes all the list of actual nodes other than itself. Initializes the
   * predecessor and successor.
   * 
   * @param the_actual_nodes is an array of all actual nodes.
   */
  private void initialNodeList(final Object[] the_actual_nodes)
  {
    my_successor = this;
    my_predecessor = this;
    int i = 0;
    final int length = the_actual_nodes.length;
    if (length > 1)
    {
      for (; i != length; ++i)
      {
        if ((int) the_actual_nodes[i] == this.my_id)
        {
          break;
        }
      }
      try
      {
        // gets the predecessor.
        my_predecessor =
            (Node) getRegistry().lookup(computePredecessor(i, length, the_actual_nodes));
        // gets the successor.
        my_successor =
            (Node) getRegistry().lookup(computeSuccessor(i, length, the_actual_nodes));
        final int first = i + 1;
        final int last = i + length;
        for (int j = first; j != last; ++j)
        {
          my_actual_list.add((int) the_actual_nodes[j % length]);
        }
      }
      catch (AccessException e)
      {
        
        e.printStackTrace();
      }
      catch (RemoteException e)
      {
        
        e.printStackTrace();
      }
      catch (NotBoundException e)
      {
        
        e.printStackTrace();
      }
    }
  }

  /**
   * Initializes the virtual nodes that this actual node should implement.
   * 
   * @throws RemoteException if there is a problem getting the successor id.
   */
  private void initializeVirtualNodes() throws RemoteException
  {

    int first = my_id;
    int last;
    final int n = SYSTEMSIZE;
    final int successor = my_successor.getMyID();
    if (successor == my_id)
    {
      first = 0;
      last = n;
    }
    else if (successor > my_id)
    {
      last = successor;
    }
    else
    {
      last = successor + n;
    }
    //
    // creates all the virtual nodes include itself
    for (int i = first; i != last; ++i)
    {
      final int id = i % n;
      final VirtualNode virtual_node = new VirtualNode(id, this, my_m_value);
      // creates the virtual channel for each virtual node.
      // virtual_node.initializeVChannel(my_m_value);
      my_virtual_nodes.put(virtual_node.getMyID(), virtual_node);
    }

  }

  /**
   * Initializes all the channels.
   * 
   * @param my_actual_list is the collection of all actual nodes.
   * @throws RemoteException if there is a problem completing the method call.
   */
  private void initializeChannel(final Object[] node) throws RemoteException
  {
    // initializes the channel
    for (int i = 0; i != my_m_value; ++i)
    {
      final ActualChannel channel =
          new ActualChannel(this, i, my_m_value, node, getRegistry());
      my_channels.put(i, channel);
    }
  }

  @Override
  public void initializeDataTransfer_Delete()
  {
    try
    {
      if (my_successor.getMyID() == my_id)
      {
        getRegistry().unbind("node_id_" + my_id);
        System.out.println("shutting down node " + my_id);
        my_out.out("shutting down node " + my_id);
        new Thread(new ShutDownThread(this)).start();
      }
      else
      {
        final Message message =
            new Message(my_id, my_predecessor.getMyID(), null, null, MessageType.DELETE_NODE);
        loadVirtualNodeData_delete(message);
        initializeDataTransfer(message);
      }
    }
    catch (RemoteException e)
    {
     
      e.printStackTrace();
    }
    catch (NullPointerException e)
    {
      System.out.println("error" + e);
    }
    catch (NotBoundException e)
    {
      
      e.printStackTrace();
    }
  }

  public void initializeDataTransfer_Add()
  {
    try
    {
      final Message message =
      // new Message(the_origin, the_destination, key_string, my_payload,
      // the_type);
          new Message(my_id, my_predecessor.getMyID(), null, null, MessageType.ADD_NODE);
      initializeDataTransfer(message);
    }
    catch (final RemoteException e)
    {
      
      e.printStackTrace();
    }
    catch (final NullPointerException e)
    {
      System.out.println("error" + e);
    }
  }

  /**
   * Sends a message to its destination.
   * 
   * @param message .
   * @throws RemoteException
   */
  public void initializeDataTransfer(final Message message) throws RemoteException
  {
    System.out.println("Node " + getMyID() + "is sending data transfer message to node: " +
                       message.getDestination() + " message type: " + message.my_type);
    my_out.out("Node " + getMyID() + "is sending data transfer message to node: " +
               message.getDestination() + " message type: " + message.my_type);
    send_message(message);
  }

  
  // for each virtual node that the new node take, it should also update the
  // channel
  public boolean updateNetworkChannels()
  {
    // key is the id of each virtual node.
    Iterator<Integer> it = my_virtual_nodes.keySet().iterator();
    while (it.hasNext())
    {
      final int update_node = it.next();
      for (int i = 0; i != my_m_value; i++)
      {
        final int offset = 1 << i;
        final int destination = (update_node - offset + SYSTEMSIZE) % SYSTEMSIZE;
        if (!containVirtual(destination))
        {
          final Message message =
              new Message(my_id, destination, null, null, MessageType.UPDATE_CHANNELS);
          // send_message(message);
          
          send_message(message);
        }
      }
    }
    System.out.println("Node " + my_id + " finished updating network channels");
    return true;
  }

  public void updateActualNodes()
  {
    try
    {
      my_predecessor.reinitialize();
      my_successor.reinitialize();
    }
    catch (RemoteException e)
    {
      
      e.printStackTrace();
    }
  }

  /**
   * Puts the node and channel information into a log file.
   * 
   * @throws RemoteException if there are problems getting the predecessor's id.
   */
  public void log() throws RemoteException
  {
    my_out.out("The node id " + my_id + " is added to the system with m = " + my_m_value);
    my_out.out("The successor node id is " + my_successor.getMyID());
    my_out.out("The predecessor node id is " + my_predecessor.getMyID());
    my_out.out("The virtual nodes that it simulates:");
    my_out.printTable(my_virtual_nodes.keySet().toArray());
    // prints out the channel information.
    for (int i = 0; i != my_channels.size(); ++i)
    {
      my_channels.get(i).printLog(my_out);
    }
  }

  @Override
  public void processRequest(final Message the_request)
  {
    final int destination = the_request.getTarget();
    if (the_request.getType() == MessageType.GET_VALUE)
    {
      final Serializable value =
          my_virtual_nodes.get(destination).get(the_request.getKeyString());
      the_request.setValue(value);
      System.out.println("GET: gets the value (" + the_request.getValue() +
                         ") for the request " + the_request.getMyId());
      my_out.out("GET: gets the value (" + the_request.getValue() + ") for the request " +
                 the_request.getMyId());
    }
    else if (the_request.getType() == MessageType.PUT_VALUE)
    {
      my_virtual_nodes.get(destination)
          .put(the_request.getKeyString(), the_request.getValue());
      System.out.println("PUT: puts the key (" + the_request.getKeyString() + ")----(" +
                         the_request.getValue() + " to the destination :" +
                         my_virtual_nodes.get(destination).getMyID());
      my_out.out("PUT: puts the key (" + the_request.getKeyString() + ")----(" +
                 the_request.getValue() + " to the destination :" +
                 my_virtual_nodes.get(destination).getMyID());
    }
  }

  
  // The client calls this method to ask for the value of the giving key.
  // Node generate a message for this request and sends it around.

  @Override
  public Serializable get(final String the_key) throws RemoteException
  {
    
    Message result = null;
    try
    {
      if (the_key == null)
      {
        throw new DataFormatException();
      }

      int hash = the_key.hashCode();
      if (hash < 0)
      {
        hash *= -1;
      }
      final int destination = hash % SYSTEMSIZE;
      final Message the_request =
          new Message(my_id, destination, the_key, null, MessageType.GET_VALUE);
      my_out.out("Get GET request from client with key: " + the_key);
      System.out.println("Get GET request from client with key: " + the_key);
      result = getResponse(the_request);
    }
    catch (final DataFormatException e)
    {
      System.out.println("The request is empty.");
    }
    return result;
  }

  @Override
  public Serializable put(final String the_key, final Serializable the_value)
      throws RemoteException
  {
    
    Message result = null;
    try
    {
      if (the_key == null)
      {
        throw new DataFormatException();
      }

      int hash = the_key.hashCode();
      if (hash < 0)
      {
        hash *= -1;
      }
      final int destination = hash % SYSTEMSIZE;
      final Message the_request =
          new Message(my_id, destination, the_key, the_value, MessageType.PUT_VALUE);
      my_out.out("Get PUT request from client with key: " + the_key + "---- value: " +
                 the_value);
      System.out.println("Get PUT request from client with key: " + the_key + "---- value: " +
                         the_value);
      result = getResponse(the_request);
    }
    catch (final DataFormatException e)
    {
      System.out.println("The request is empty.");
    }
    return result;
  }

  /**
   * Client would call this method to process a request and get the result.
   * 
   * @param the_request is the request.
   * @return
   * @throws RemoteException
   */
  private Message getResponse(final Message the_request) throws RemoteException
  {

    // checks if the destination
    if (containVirtual(the_request.getDestination()))
    {
      this.processRequest(the_request);
      return the_request;
    }

    System.out.println("Sends the request: " + the_request.getMyId() +
                       " to the destination: " + the_request.getTarget());
    my_out.out("Sends the request: " + the_request.getMyId() + " to the destination: " +
               the_request.getTarget());

    send_message(the_request);
    Message the_result;

    synchronized (my_message_monitor)
    {
      while (my_results_table.get(the_request.getMyId()) == null)
      {
        try
        {
          
          my_message_monitor.wait(100000);
        }
        catch (InterruptedException e)
        {
          
          e.printStackTrace();
        }
      }
      the_result = my_results_table.get(the_request.getMyId());
      my_out.out("Gets the response for request: " + the_request.getMyId() +
                 " and sending back to client.");
      my_results_table.remove(the_result.getMyId());
    }
    return the_result;
  }

  @Override
  public void send_message(final Message the_request)
  {
    try
    {
      final Node destination = nextHop(the_request);
      System.out.println("Node " + getMyID() + " sending message " + the_request.getMyId() +
                         " to next hop: " + destination.getMyID() +
                         " and the final destination is " + the_request.getDestination());
      my_out.out("Node " + getMyID() + " sending message " + the_request.getMyId() +
                 " to next hop: " + destination.getMyID() + " and the final destination is " +
                 the_request.getDestination());
      destination.process_message(the_request);
    }
    catch (RemoteException e)
    {
     
      e.printStackTrace();
    }
  }

  public void updateResultTable(final Message the_message)
  {
    my_results_table.put(the_message.getMyId(), the_message);
  }

  public void process_message(final Message the_message)
  {

    System.out.println("Node id " + my_id + " --- Got a message from somewhere... ");
    my_out.out("Node id " + my_id  + " --- Got the message " + the_message.getMyId());
    my_thrd_pool.execute(new MessageProcessor(my_results_table, the_message, this,
                                             my_message_monitor));
  }

 
  public void reinitializeChannels()
  {
    System.out.println("Node " + my_id + " is reinitializing its channels");
    try
    {
      reinitialize();
    }
    catch (RemoteException e)
    {
            e.printStackTrace();
    }
  }
/**
 * 
 * @param the_message .
 */
  public void loadVirtualNodeData_delete(final Message the_message)
  {
    final HashMap<Integer, HashMap<String, Serializable>> virtual_node_data =
        new HashMap<Integer, HashMap<String, Serializable>>();
    final int size = my_virtual_nodes.size();
    for (int i = my_id; i != my_id + size; ++i)
    {
      virtual_node_data.put(i % SYSTEMSIZE, (HashMap<String, Serializable>) my_virtual_nodes
          .get(i % SYSTEMSIZE).getData());
    }
    the_message.setValue(virtual_node_data);

    System.out.println("node " + my_id + "loaded the virtual node data into a message!");
  }

  
  /**
   * 
   */
  public void loadVirtualNodeData(final Message the_message)
  {
    System.out.println("this node was simulating nodes: " +
                       (my_virtual_nodes.keySet()).toString());

    int new_node_id = the_message.getOrigin();
    if (the_message.getType().equals(MessageType.ADD_NODE))
    {
      new_node_id = the_message.getOrigin();
    }
    else if (the_message.getType().equals(MessageType.DELETE_NODE))
    {
      new_node_id = the_message.getTarget();
    }

    final HashMap<Integer, HashMap<String, Serializable>> virtual_node_data =
        new HashMap<Integer, HashMap<String, Serializable>>();

    for (int i = 0; i < my_virtual_nodes.size(); i++)
    {
      final int index = (new_node_id + i) % SYSTEMSIZE;
      if (my_virtual_nodes.get(index) == null || index == my_id)
      {
        break;
      }
      else
      {
        final HashMap<String, Serializable> data =
            (HashMap<String, Serializable>) my_virtual_nodes.get(index).getData();
        virtual_node_data.put(index, data);
      }
    }

    the_message.setValue(virtual_node_data);

    System.out.println("node " + my_id + "loaded the virtual node data into a message!");
  }
/**
 * 
 */
  public void deleteTransferedVirtualNodes(final Message the_message)
  {
    final int new_node_id = the_message.getOrigin();
    final int size = my_virtual_nodes.size();
    for (int i = 0; i < size; i++)
    {
      final int index = (new_node_id + i) % SYSTEMSIZE;
      if (my_virtual_nodes.get(index) == null || index == my_id)
      {
        break;
      }
      else
      {
        my_virtual_nodes.remove(index);
      }
    }
    System.out.println("this node now simulates nodes: " +
                       (my_virtual_nodes.keySet()).toString());
    my_out.out("Updated the virtual nodes, now simulating nodes: ");
    my_out.printTable(my_virtual_nodes.keySet().toArray());
  }
/**
 * 
 */
  // Unpackages the virtual node data
  public void unloadVirtualNodeData(final Message the_message)
  {

    final HashMap<Integer, HashMap<String, Serializable>> nodes =
        (HashMap<Integer, HashMap<String, Serializable>>) the_message.getValue();
    final Iterator<Integer> it = nodes.keySet().iterator();

    while (it.hasNext())
    {
      final Integer a_virt_node_id = it.next();
      final HashMap<String, Serializable> data = nodes.get(a_virt_node_id);
      final VirtualNode v_node = new VirtualNode(a_virt_node_id, this, data, my_m_value);
      my_virtual_nodes.put(a_virt_node_id, v_node);
    }
    System.out.println("the node: " + my_id + " now simulates nodes: " +
                       (my_virtual_nodes.keySet()).toString());
    my_out.out("Updated the virtual nodes, now simulating nodes: ");
    my_out.printTable(my_virtual_nodes.keySet().toArray());
    // send acknowledgment message to whoever sent these nodes so it can delete
    // them...
    try
    {
      if (the_message.getType().equals(MessageType.ADD_NODE))
      {
        System.out.println("the node: " + my_id +
                           " is now sending an add acknowledgement to " +
                           my_predecessor.getMyID());
        final Message message =
            new Message(my_id, my_predecessor.getMyID(), null, null, MessageType.ADD_NODE_ACK);
        send_message(message);
        System.out.println("Node " + getMyID() + " System initialized.");
      }
      else if (the_message.getType().equals(MessageType.DELETE_NODE))
      {
        System.out.println("the node: " + my_id +
                           " is now sending an delete acknowledgement to " +
                           the_message.getOrigin());
        final Message message =
            new Message(my_id, the_message.getOrigin(), null, null,
                        MessageType.DELETE_NODE_ACK);
        send_message(message);
      }
    }
    catch (RemoteException e)
    {
      
      e.printStackTrace();
    }
  }

  @Override
  public boolean containVirtual(final int the_virtual)
  {
    return my_virtual_nodes.containsKey(the_virtual);
  }

  /**
   * Gets the next hop id from the finger table. In this method, it first probe
   * the channel of the intermediate node - virtual node-then update the new
   * intermediate node in the request, then the actual node probes its channel
   * and gives proper neighbor actual node.
   * 
   * @param the_request is the request.
   * @return the actual node of the next hop.
   * @throws RemoteException if there is a problem happened during the remote
   *           call.
   */
  public Node nextHop(final Message the_request) throws RemoteException
  {
    Node result = null;
    final VirtualNode vnode = my_virtual_nodes.get(the_request.getIntermediate());
    if (vnode == null)
    {
      // System.err.println("Error happened in next Hop");
      result = probeChannels(the_request.getIntermediate());
    }
    else
    {
      final int next = vnode.probeChannels(the_request.getDestination());

      // System.out.println("My intermediate node is " +
      // the_request.getIntermediate());
      // update the intermediate node id in the request.
      the_request.setIntermediate(next);
      result = probeChannels(next);
      int n = result.getMyID();
      System.out.println("final output node id is " + n);
    }
    return result;
  }

  /**
   * Probes for the appropriate neighboring node to reach key node.
   * 
   * @param the_node is the destination node.
   * @return the actual node.
   */
  public Node probeChannels(final int the_node) throws RemoteException
  {
    // where we will check all the channels and return the neighbor we need
    // to pass message to.
    Node node = null;
    for (int i = my_m_value - 1; i >= 0; --i)
    {
      if (my_channels.get(i).isInInterval(the_node))
      {
        node = my_channels.get(i).getActualNode();
        if (node.getMyID() == this.my_id)
        {
          // return the successor.
          node = this.my_successor;
        }
        
        break;
      }
    }
    if (node == null)
    {
      node = this;
    }
    return node;
  }

  @Override
  public void initializeShutDown(final Message my_message) throws RemoteException
  {
    // Use later to make the node leave the network.
    // registry.unbind("node_id_" + node_id);
    // UnicastRemoteObject.unexportObject(my_node, true);

    try
    {
      this.getRegistry().unbind("node_id_" + my_id);
      System.out.println("unbinding node " + my_id);

      System.out.println("Updating channels of neighbors pointing to the new node");
      updateNetworkChannels();

      System.out.println("Updating Successor & Predecessor actual nodes..........");
      updateActualNodes();

      System.out.println("shutting down node " + my_id);
      UnicastRemoteObject.unexportObject(this, true);
      System.exit(0);

    }
    catch (NotBoundException e)
    {
      
      e.printStackTrace();
    }
  }

  public int getMyID() throws RemoteException
  {
    return my_id;
  }

  @Override
  public int getMValue() throws RemoteException

  {

    return my_m_value;

  }

  public Registry getRegistry() throws RemoteException

  {

    return my_registry;

  }

}
